#pragma once

#include <memory>
#include <vector>
#include <iostream>
#include <string>
#include <unordered_map>
#include <mutex>
#include <muduo/net/TcpClient.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/EventLoop.h>

#include "abstract.hpp"
#include "fields.hpp"
#include "log.hpp"
#include "json.hpp"
#include "message.hpp"

namespace JsonMuduoRPC
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;

        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf) {}

        virtual size_t readableBytes() override
        {
            return _buf->readableBytes();
        }
        virtual int32_t peekInt32() override
        {
            // 尝试取出四字节数据，并作网络字节序转换 网络->主机
            return _buf->peekInt32();
        }
        virtual void retrieveInt32() override
        {
            _buf->retrieveInt32();
        }
        virtual int32_t readInt32() override
        {
            return _buf->readInt32();
        }
        virtual std::string retrieveAsString(size_t len) override
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
        /*
            |--length--|--MType--|--IDLength--|--MID--|--Body--|
        */
    public:
        using ptr = std::shared_ptr<LVProtocol>;

        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if (buf->readableBytes() < _lenFields_length)
                return false;

            int32_t total_len = buf->peekInt32();
            if (buf->readableBytes() < (total_len + _lenFields_length))
                return false;

            return true;
        }

        // 处理收到的消息
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            // 调用该函数时，默认已经收到了一条完整的消息

            int32_t total_len = buf->readInt32();
            MType mType = (MType)buf->readInt32();
            int32_t id_Len = buf->readInt32();
            int32_t body_len = total_len - _mTypeFields_length - _idLenFields_length - id_Len;

            std::string id = buf->retrieveAsString(id_Len);
            std::string body = buf->retrieveAsString(body_len);

            msg = MessageFactory::create(mType);
            if (msg.get() == nullptr)
            {
                LOG(ERROR, "无效的类型");
                return false;
            }

            bool ret = msg->unSerialize(body);
            if (ret == false)
            {
                LOG(ERROR, "消息反序列化失败");
                return false;
            }

            msg->setId(id);
            msg->setMType(mType);

            return true;
        }

        // 发消息前要根据LV协议，对消息序列化
        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            // |--length--|--MType--|--IDLength--|--MID--|--Body--|

            std::string body;
            msg->serialize(body);
            std::string id = msg->getId();
            auto mType = htonl((int32_t)msg->getMType());

            int32_t id_len = htonl(id.size());
            int32_t h_total_len = _mTypeFields_length + _idLenFields_length + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);

            std::string result;
            result.append((char *)&n_total_len, _lenFields_length);
            result.append((char *)&mType, _mTypeFields_length);
            result.append((char *)&id_len, _idLenFields_length);
            result.append(id);
            result.append(body);

            return result;
        }

    private:
        const int32_t _lenFields_length = 4;
        const int32_t _mTypeFields_length = 4;
        const int32_t _idLenFields_length = 4;
    };

    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;

        MuduoConnection(const muduo::net::TcpConnectionPtr &con, const BaseProtocol::ptr &protocol)
            : _con(con), _protocol(protocol)
        {
        }

        virtual bool send(const BaseMessage::ptr &msg) override
        {
            std::string body = _protocol->serialize(msg);
            _con->send(body);

            return true;
        }
        virtual void shutdown() override
        {
            _con->shutdown();
        }
        virtual bool connected() override
        {
            return _con->connected();
        }

    private:
        muduo::net::TcpConnectionPtr _con;
        BaseProtocol::ptr _protocol;
    };

    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;

        MuduoServer(uint16_t port)
            : _muduo_server(&_event_loop,
                            muduo::net::InetAddress("0.0.0.0", port),
                            "MuduoServer",
                            muduo::net::TcpServer::kReusePort), // 启用端口复用
              _protocol(ProtocolFactory::create())
        {
        }

        virtual void start() override
        {
            _muduo_server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _muduo_server.setMessageCallback(std::bind(&MuduoServer::onMessage,
                                                       this,
                                                       std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _muduo_server.start();
            _event_loop.loop();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &con)
        {
            if (con->connected())
            {
                LOG(INFO, "连接建立");

                // 获得一个新连接，就要对TcpConnectionPtr封装
                // 建立TcpConnectionPtr和BaseConnectionPtr的映射
                BaseConnection::ptr base_con_ptr = ConnectionFactory::create(con, _protocol);
                {
                    // 要对临界资源作修改，因此要加锁
                    std::unique_lock<std::mutex> lock(_mutex);

                    _con_map.insert(std::make_pair(con, base_con_ptr));
                }
                if (_cb_connection)
                    _cb_connection(base_con_ptr);
            }
            else
            {
                LOG(INFO, "连接断开");

                BaseConnection::ptr base_con_ptr;
                {
                    std::unique_lock<std::mutex> lock(_mutex);

                    auto it = _con_map.find(con);
                    if (it == _con_map.end())
                    {
                        LOG(WARNING, "未知连接");
                        return;
                    }
                    base_con_ptr = it->second;
                    _con_map.erase(con); // 连接断开，删除映射关系
                }
                if (_cb_close)
                    _cb_close(base_con_ptr);
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &con, muduo::net::Buffer *buffer, muduo::Timestamp)
        {
            LOG(INFO, "连接有消息到来，开始处理...");

            BaseBuffer::ptr base_buf_ptr = BufferFactory::create(buffer);
            while (true)
            {
                // 先检验消息是否能够处理
                if (_protocol->canProcessed(base_buf_ptr) == false)
                {
                    if (base_buf_ptr->readableBytes() > max_data_size)
                    {
                        LOG(FATAL, "消息过长，连接中断");
                        con->shutdown();
                        return;
                    }

                    //LOG(DEBUG, "消息长度不足");
                    return;
                }

                // 通过协议，处理消息
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf_ptr, msg);
                if (ret == false)
                {
                    LOG(ERROR, "缓冲区中数据错误");
                    con->shutdown();
                    return;
                }

                // 将处理完后的消息，交给回调进行处理
                //LOG(INFO, "消息处理完毕，执行回调");
                BaseConnection::ptr base_con_ptr;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _con_map.find(con);
                    if (it == _con_map.end())
                    {
                        LOG(ERROR, "无效的连接");
                        con->shutdown();
                        return;
                    }

                    base_con_ptr = it->second;
                }
                if (_cb_message)
                    _cb_message(base_con_ptr, msg);
            }
        }

    private:
        const ssize_t max_data_size = (1 << 16); // 最大消息大长度 64k

        // 注意，TcpServer的初始化依赖于EventLoop
        // 所以，_event_loop要放在_muduo_server之上
        muduo::net::EventLoop _event_loop;
        muduo::net::TcpServer _muduo_server;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _con_map;
        std::mutex _mutex;
        BaseProtocol::ptr _protocol;
    };

    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;

        MuduoClient(const std::string &serverIP, const uint16_t serverPort)
            : _eventLoop(_threadLoop.startLoop()),
              _client(_eventLoop, muduo::net::InetAddress(serverIP, serverPort), "DictClient"),
              _downLatch(1), // 初始化计数器为1，当成功建立连接，计数器减为0，此时可以发送数据
              _protocol(ProtocolFactory::create())
        {
        }

        virtual void connect() override
        {
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage,
                                                 this,
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 等待连接至服务器
            _client.connect();
            _downLatch.wait(); // 等待计数器减为0
        }
        virtual bool connected() override
        {
            return (_con && _con->connected());
        }
        virtual BaseConnection::ptr connection() override
        {
            return _con;
        }
        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (connected() == false)
            {
                LOG(ERROR, "连接断开，发送失败！！");
                return false;
            }
            _con->send(msg);
            return true;
        }
        virtual void shutdown() override
        {
            _client.disconnect();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &con)
        {
            if (con->connected())
            {
                LOG(INFO, "连接建立");
                _con = ConnectionFactory::create(con, _protocol);
                _downLatch.countDown(); // 计数器--
            }
            else
            {
                LOG(INFO, "连接断开");
                _con.reset();
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &con, muduo::net::Buffer *buffer, muduo::Timestamp)
        {
            LOG(INFO, "连接有消息到来，开始处理...");

            BaseBuffer::ptr base_buf_ptr = BufferFactory::create(buffer);
            while (true)
            {
                // 先检验消息是否能够处理
                if (_protocol->canProcessed(base_buf_ptr) == false)
                {
                    if (base_buf_ptr->readableBytes() > max_data_size)
                    {
                        LOG(FATAL, "消息过长，连接中断");
                        con->shutdown();
                        return;
                    }

                    //LOG(DEBUG, "消息长度不足");
                    return;
                }

                // 通过协议，处理消息
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf_ptr, msg);
                if (ret == false)
                {
                    LOG(ERROR, "缓冲区中数据错误");
                    _con->shutdown();
                    return;
                }

                // 将处理完后的消息，交给回调进行处理
                if (_cb_message)
                    _cb_message(_con, msg);
            }
        }

    protected:
        const ssize_t max_data_size = (1 << 16); // 最大消息大长度 64k

        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _con;

        // 需要启动loop()循环开始事件监控，但是这是一个死循环，如果直接放到成员函数中，就会导致程序无法继续运行
        // 可以利用一个线程来loop
        muduo::net::EventLoopThread _threadLoop;
        muduo::net::EventLoop *_eventLoop;
        muduo::net::TcpClient _client;
        muduo::CountDownLatch _downLatch;
    };

    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}
